package org.ykhl.mq.commons;

import org.apache.commons.codec.digest.DigestUtils;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.*;
import java.security.KeyStore;
import java.security.KeyStoreException;
import java.security.NoSuchAlgorithmException;
import java.security.cert.Certificate;
import java.security.cert.CertificateException;
import java.security.cert.X509Certificate;
import java.util.*;
import java.util.concurrent.*;
import java.util.regex.Pattern;

/**
 * Created by zkf on 2015/11/23.
 */
public class Utils {
    private static final Logger LOG = LoggerFactory.getLogger(Utils.class);
    private static Properties producerPro = new Properties();
    private static Properties consumerPro = new Properties();
    private Utils(){}


    public void setProducerPro(Properties producerPro) {
        this.producerPro = producerPro;
    }

    public void setConsumerPro(Properties consumerPro) {
        this.consumerPro = consumerPro;
    }
    /**
     *
     *从keystore证书文件中获取authId.
     * @param filePath keystore证书文件的绝对路径
     * @return
     * @throws IOException
     * @throws CertificateException
     * @throws NoSuchAlgorithmException
     * @throws KeyStoreException
     */
    public static String getAuthIdFromKeyStoreUnchecked(String filePath) {
        try {
            return getAuthIdFromKeyStore(new FileInputStream(new File(filePath)));
        } catch (RuntimeException re) {
            throw re;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }

    /**
     * 从keystore证书文件中获取authId.
     *
     * @param filePath keystore证书文件的绝对路径
     * @return
     */
    public static String getAuthIdFromKeyStore(String filePath)
            throws IOException, CertificateException,
            NoSuchAlgorithmException, KeyStoreException {
        return getAuthIdFromKeyStore(new FileInputStream(new File(filePath)));
    }

    /**
     * 从keystore证书文件中获取authId.
     *
     * @param stream
     * @return
     */
    public static  String getAuthIdFromKeyStoreUnchecked(InputStream stream) {
        try {
            return getAuthIdFromKeyStore(stream);
        } catch (RuntimeException re) {
            throw re;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 从keystore证书文件中获取authId.
     *
     * @param stream
     * @return
     * @throws IOException
     * @throws KeyStoreException
     * @throws CertificateException
     * @throws NoSuchAlgorithmException
     */
    public static String getAuthIdFromKeyStore(InputStream stream)
            throws IOException, KeyStoreException,
            CertificateException, NoSuchAlgorithmException {
        try {
            KeyStore keyStore = KeyStore.getInstance(Constants.KEYSTORE_TYPE);
            keyStore.load(stream, Constants.KEYSTORE_PASSWD);
            Enumeration<String> aliases = keyStore.aliases();
            while (aliases.hasMoreElements()) {
                String alias = aliases.nextElement();
                if (!Constants.KEYSTORE_ALIAS_NAME.equals(alias))
                    continue;

                Certificate cert = keyStore.getCertificate(alias);
                if (!(cert instanceof X509Certificate))
                    throw new CertificateException("Certificate is not X509 type!");

                String dn = ((X509Certificate) cert).getSubjectDN().getName();
                int st = dn.indexOf('-'), end = dn.indexOf(',');
                if (st == -1 || end == -1 || end <= st)
                    throw new CertificateException("Certificate has wrong domain name!");

                return dn.substring(st + 1, end).trim();
            }

            throw new KeyStoreException("No certificate named " + Constants.KEYSTORE_ALIAS_NAME + " was found");
        } finally {
            stream.close();
        }
    }

    /**
     * 创建一个可以阻塞的ExecutorService, 当任务队列到达queueSize时, 使用CallerRunsPolicy.
     *
     * ThreadPoolExecutor.AbortPolicy()              ---   抛出java.util.concurrent.RejectedExecutionException异常
     *ThreadPoolExecutor.CallerRunsPolicy()       ---   重试添加当前的任务，他会自动重复调用execute()方法
     *ThreadPoolExecutor.DiscardOldestPolicy()   ---   抛弃旧的任务
     *ThreadPoolExecutor.DiscardPolicy()            ---   抛弃当前的任务
     *ArrayBlockingQueue 有界队列 可以防止资源耗尽的情况发生
     * @param threadCnt 线程池的线程数量
     * @param queueSize 任务队列的最大长度
     * @return
     */
    public static ExecutorService newBlockingFixedThreadPool(int threadCnt, int queueSize) {
        return new ThreadPoolExecutor(
                threadCnt, threadCnt, 0L, TimeUnit.MICROSECONDS,
                new ArrayBlockingQueue<Runnable>(queueSize, true),
                new ThreadPoolExecutor.CallerRunsPolicy());
    }

    /**
     * 创建一个可以阻塞的ExecutorService, 当任务队列到达queueSize时, 使用CallerRunsPolicy.
     *
     * @param threadCnt 线程池的线程数量
     * @param queueSize 任务队列的最大长度
     * @return
     */
    public static ExecutorService newBlockingFixedThreadPool(int threadCnt, int queueSize,
                                                             ThreadFactory threadFactory) {
        return new ThreadPoolExecutor(
                threadCnt, threadCnt, 0L, TimeUnit.MICROSECONDS,
                new ArrayBlockingQueue<Runnable>(queueSize, true), threadFactory,
                new ThreadPoolExecutor.CallerRunsPolicy());
    }

    public static String genMd5(String body){
        return DigestUtils.md5Hex(body.getBytes(Constants.CHARSET));
    }

    private static final Boolean USE_SECURITY_DEFAULT = false;
    public static Properties genKafkaProducerProps(String serverAddress){
        return genKafkaProducerProps(serverAddress, USE_SECURITY_DEFAULT);
    }
    public static  Properties genKafkaConsumerProps(String serverAddress, String groupName){
        return genKafkaConsumerProps(serverAddress, groupName, USE_SECURITY_DEFAULT);
    }

    public static Properties genKafkaProducerProps(String serverAddress, Boolean useSsl){
        return genKafkaProducerProps(serverAddress, useSsl, null);
    }

    public static Properties genKafkaProducerProps(String serverAddress, Boolean useSsl, String keystorePath){

        Properties  props = new Properties();
        try {
            props.load(Utils.class.getClassLoader().getResourceAsStream("kafka/ykhlpro.properties"));
            Set<Object> s = props.keySet();
            for (Object ob : s) {
                Object value = props.getProperty(ob.toString());

                props.put(ob.toString(), value);
            }
            props.put("bootstrap.servers", serverAddress);
            props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            if (useSsl == null ? USE_SECURITY_DEFAULT : useSsl)
                putKafkaSecurityConfig(props, keystorePath);

        }catch (Exception e){
            e.printStackTrace();

        }
        return props;
    }

    public static Properties genKafkaConsumerProps(String serverAddress, String groupName, Boolean useSsl){
        return genKafkaConsumerProps(serverAddress, groupName, useSsl, null);
    }

    public static Properties genKafkaConsumerProps(String serverAddress, String groupName, Boolean useSsl, String keystorePath){

        Properties  props = new Properties();
        try {
           props.load(Utils.class.getClassLoader().getResourceAsStream("kafka/ykhlcon.properties"));
           Set<Object> s =  props.keySet();

            for (Object ob:s){
                Object value = props.getProperty(ob.toString());
                props.put(ob.toString(),value);
            }
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, serverAddress);
            props.put(ConsumerConfig.GROUP_ID_CONFIG, groupName);
            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

            if (useSsl == null ? USE_SECURITY_DEFAULT : useSsl)
                putKafkaSecurityConfig(props, keystorePath);

        }catch (Exception e){
            e.printStackTrace();
        }
             return  props;
    }

    public static Pattern genKafkaSubscribePatternByPrefix(String prefix){
        return Pattern.compile(prefix + ".*");
    }

    private static void putKafkaSecurityConfig(Properties props, String keystorePath) {
        if (keystorePath == null || keystorePath.isEmpty())
            keystorePath = Constants.KAFKA_SECURITY_KEYSTORE_LOCATION;

        props.put("security.protocol", "SSL");
        props.put("ssl.truststore.password", "123456");
        props.put("ssl.truststore.type", "JKS");
        props.put("ssl.truststore.location", keystorePath);

        props.put("ssl.keystore.password", "123456");
        props.put("ssl.keystore.type", "JKS");
        props.put("ssl.key.password", "123456");

        props.put("ssl.keystore.location", keystorePath);
    }
}
